from confluent_kafka import Consumer, KafkaException
import sys,os

class KafkaConsumerEcho:
    def __init__(self, bootstrap_servers, group_id, topic):
        # 配置消费者
        self.consumer = Consumer({
            'bootstrap.servers': bootstrap_servers,
            'group.id': group_id,
            'auto.offset.reset': 'earliest'  # 从最早的偏移量开始消费
        })
        self.topic = topic

    def consume_messages(self):
        # 订阅指定主题
        self.consumer.subscribe([self.topic])

        print(f"Subscribed to topic: {self.topic}")

        try:
            while True:
                # 消费消息
                msg = self.consumer.poll(timeout=1.0)

                if msg is None:
                    continue

                if msg.error():
                    raise KafkaException(msg.error())

                # 输出消息
                print(f"Received message: {msg.value().decode('utf-8')} (key: {msg.key()})")

        except KeyboardInterrupt:
            print("Stopping consumer...")
        finally:
            # 确保关闭消费者
            self.consumer.close()

# 使用示例
if __name__ == "__main__":
    if len(sys.argv) != 4:
        print(f"Usage: {sys.argv[0]} <bootstrap_servers> <group_id> <topic>")
        sys.exit(1)
    try:
        bootstrap_servers = sys.argv[1]
        group_id = sys.argv[2]
        topic = sys.argv[3]

        consumer = KafkaConsumerEcho(bootstrap_servers, group_id, topic)
        consumer.consume_messages()
    except KafkaException as e:
        print(f"Kafka error: {e}")
    except Exception as e:
        print(f"An error occurred: {e}")
    os.system("pause")
